package cn.easy.jacob.rocketmq;

import cn.easy.jacob.service.RocketMQService;
import org.apache.rocketmq.client.consumer.MQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class MessageListener implements MessageListenerConcurrently {
    private static final Logger LOGGER = LoggerFactory.getLogger(MQPushConsumer.class);

    @Autowired
    private RocketMQService rocketMQService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt msgTimes = null;
        try {
            for (MessageExt messageExt : msgs) {
                msgTimes = messageExt;
                LOGGER.info("MQ：消费者接收新信息: {} {} {} {} {}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getTags(), messageExt.getKeys(), "正在转换文件");

                //发送消息
                rocketMQService.sendMessage(messageExt);
                LOGGER.info("MQ：消费者接收新信息: {} {} {} {} {}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getTags(), messageExt.getKeys(), "转换完成");

            }
        } catch (Exception e) {
            LOGGER.error("MQ：消费者接收新信息: {}", e.getMessage());
            if (msgTimes.getReconsumeTimes() == 5){
                LOGGER.error("MQ：消费者接收新信息: {}", msgs.get(0).getMsgId(), "已失败，不再重试");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            //消费失败 稍后再试
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        //消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
